package com.test.chat.handler;

import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.test.chat.message.MessageConstants;
import com.test.chat.message.TextMessage;
import com.test.chat.session.LocalSessionManager;
import com.test.chat.topic.TopicManager;
import com.test.chat.utils.SerializationUtil;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import io.netty.util.AttributeKey;

public class MqttHandler extends ChannelHandlerAdapter {
	
	private static final  AttributeKey<String> ATTR_USERNAME = AttributeKey.valueOf("username");
	private LocalSessionManager localSessionManager = new LocalSessionManager();
	private TopicManager topicManager = new TopicManager();
	private ExecutorService executorService = Executors.newFixedThreadPool(100);
	
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
	}
	
	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		Channel channel = ctx.channel();
		MqttMessage mqttMessage = (MqttMessage) msg;
		switch (mqttMessage.fixedHeader().messageType()) {
		case CONNECT:
			processConnectMessage(channel, (MqttConnectMessage) msg);
			break;
		case CONNACK:
			break;
		case SUBSCRIBE:
			processSubscribeMessage(channel, (MqttSubscribeMessage) msg);
			break;
		case UNSUBSCRIBE:
			break;
		case SUBACK:
			break;
		case UNSUBACK:
			break;
		case PUBACK:
			break;
		case PUBREC:
			break;
		case PUBCOMP:
			break;
		case PUBREL:
			break;
		case PUBLISH:
			processPublishMessage(channel, (MqttPublishMessage) msg);
			break;
		case PINGREQ:
			break;
		case PINGRESP:
			break;
		case DISCONNECT:
			break;
		}
	}
	
	private void processConnectMessage(Channel channel, MqttConnectMessage message) {
		MqttConnectPayload payload = message.payload();
		String username = payload.userName();
		String password = payload.password();
		MqttConnAckVariableHeader connAckVariableHeader = new MqttConnAckVariableHeader(
				MqttConnectReturnCode.CONNECTION_ACCEPTED);
		
		channel.attr(ATTR_USERNAME);
		localSessionManager.putSession(username, channel);
		channel.attr(ATTR_USERNAME).set(username);
		
		MqttConnAckMessage connAckMessage = new MqttConnAckMessage(message.fixedHeader(), connAckVariableHeader);
		channel.writeAndFlush(connAckMessage);
	}
	
	private void processSubscribeMessage(Channel channel, MqttSubscribeMessage message) {
		MqttFixedHeader mqttFixedHeader = message.fixedHeader();
		MqttMessageIdVariableHeader variableHeader = message.variableHeader();
		Iterator<MqttTopicSubscription> iter = message.payload().topicSubscriptions().iterator();
		while(iter.hasNext()){
			MqttTopicSubscription topicSubscription = iter.next();
			topicManager.add(topicSubscription.topicName(), channel.attr(ATTR_USERNAME).get());
			
		}
		if(mqttFixedHeader.qosLevel() == MqttQoS.AT_LEAST_ONCE){
			MqttSubAckMessage subAckMessage = new MqttSubAckMessage(mqttFixedHeader, variableHeader, null);
			channel.writeAndFlush(subAckMessage);
		}
	}
	
	private void processPublishMessage(Channel channel, MqttPublishMessage message) {
		MqttFixedHeader mqttFixedHeader = message.fixedHeader();
		MqttPublishVariableHeader variableHeader = message.variableHeader();
		
		ByteBuf payload = message.payload();
		byte type = payload.readByte();
		ByteBuf body = payload.discardReadBytes();
		
		switch (type) {
		case MessageConstants.TEXT_MESSAGE:
			TextMessage textMessage = SerializationUtil.deserialize(body.array(), TextMessage.class);
			String to = textMessage.getTo();
			Channel clintChannel = localSessionManager.getSession(to);
			clintChannel.write(textMessage);
			break;
		case MessageConstants.IMAGE_MESSAGE:
			
			break;
		default:
			break;
		}
		
		if (mqttFixedHeader.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
			MqttMessageIdVariableHeader messageIdVariableHeader = MqttMessageIdVariableHeader.from(variableHeader.messageId());
			MqttPubAckMessage pubAckMessage = new MqttPubAckMessage(mqttFixedHeader, messageIdVariableHeader);
			channel.writeAndFlush(pubAckMessage);
		}
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		localSessionManager.removeSession("");
	}
	
	@Override
	public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
		localSessionManager.removeSession("");
		super.close(ctx, promise);
	}

}
